package com.example.demo_pc.util;


import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * @Description: 多线程处理类
 * @Author jiateng
 * @Date 2020/6/28 13:47
 */
public class MultiThreadUtils<T> {
    private final Logger logger = LoggerFactory.getLogger(this.getClass());
    private int countNum;

    public int getCountNum() {
        return countNum;
    }

    public void setCountNum(int countNum) {
        this.countNum = countNum;
    }

    public void execute(List<T> data, Task<T> task) throws Exception{
        ExecutorService executors =null;
        try {
            logger.info("开始多线程处理...");
            long startTime = System.currentTimeMillis();
            //要推送的用户总数
            int count = data.size();
            logger.info("数据总个数=>{}",count);
            //初始每个线程处理的用户数量
            //计算处理所有用户需要的线程数量
            int eveBlocks =countNum; //count / eveLength + (count % eveLength != 0 ? 1 : 0);
            logger.info("需要的线程个数=>{}",eveBlocks);
            int count_task = count / eveBlocks;
            final int eveLength = count % eveBlocks != 0 ? count_task+1 : count_task;
            logger.info("每个线程处理的个数:{}",eveLength);
            executors = Executors.newFixedThreadPool(eveBlocks);
            //线程计数器
            CountDownLatch doneSignal = new CountDownLatch(eveBlocks);
            //开启线程处理
            int doneCount = 0;
            logger.info("开启线程处理...");
            for (int page = 0; page < eveBlocks; page++) { /* blocks太大可以再细分重新调度 */
                MessageSendThread ms = new MessageSendThread(data,page + 1,eveLength,doneSignal,task);
                executors.execute(ms);
                doneCount++;
            }
            doneSignal.await();//等待所有计数器线程执行完
            long endTime = System.currentTimeMillis();
            logger.info("多线程任务处理结束,总时长(秒)=>{}",(endTime-startTime)/1000);
            logger.info("线程总个数=>{}",doneCount);
        } catch (Exception e) {
            logger.error("多线程任务处理异常=>{}",e);
        }finally {
            if (executors!=null){
                executors.shutdown();
            }
        }
    }
    private  class MessageSendThread implements Runnable{
        private Integer currentIndex;//当前索引
        private Integer rows;//处理数据条数
        private CountDownLatch doneSignal;//处理线程条数
        private List<T> data;//消息实体
        private Integer first;
        private Integer end;
        private Task<T> task;

        public MessageSendThread(List<T> data,Integer currentIndex,Integer rows, CountDownLatch doneSignal,Task task) {
            this.data = data;
            this.currentIndex = currentIndex;
            this.rows = rows;
            this.doneSignal = doneSignal;
            this.first=(currentIndex-1)*rows;
            this.end=first+rows;
            this.task=task;
        }
        //过滤出指定客户
        private List<T> filterData(List<T> data,Integer first,Integer end)throws Exception{
            logger.info("过滤指定数据....");
            return data.subList(first,end>data.size()?data.size():end);
        }

        @Override
        public void run() {
            try {
                /**
                 * ---------1.查询当前的block范围内的发送的手机号=>筛选目标客户群手机号---------
                 */
                List<T> subData = filterData(data,first,end);
                /**
                 *  ---------2.批量发送消息---------
                 *
                 */
                task.execute(subData);
            } catch (Exception e) {
                logger.error("任务处理执行异常:{}",e);
            }finally{
                doneSignal.countDown();//计数器减一
            }
        }
    }
}
